package cn.admin.remoting.impl;

import cn.admin.remoting.*;
import cn.admin.remoting.locale.LogResources;
import cn.admin.remoting.util.DIYExecutor;
import cn.admin.remoting.util.LoggerInit;
import cn.admin.remoting.util.UnsafeCast;
import org.apache.commons.lang.StringUtils;
import org.apache.mina.common.IoFuture;
import org.apache.mina.common.IoFutureListener;
import org.apache.mina.common.WriteFuture;

import java.util.concurrent.Callable;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;

import static cn.admin.remoting.locale.LogResources.*;

public class DefaultRespFuture implements ResponseFuture {

    private volatile long waitBegin;
    private volatile boolean isDone;
    private volatile ConnectionResponse connResponse;
    private final ConnectionRequest connRequest;
    private ResponseCallback respCallback;
    private ScheduledFuture<?> timeoutFuture;
    private IoFutureListener writeFutureListener;

    DefaultRespFuture(ConnectionRequest connRequest) {
        this.connRequest = connRequest;
        this.writeFutureListener = new WFL();
    }

    @Override
    public long requestId() {
        return this.connRequest.getId();
    }

    @Override
    public boolean isDone() {
        return this.isDone;
    }

    @Override
    public Object get() throws TimeoutException, RemotingException, InterruptedException {
        synchronized (this) {
            while (!this.isDone) {
                this.wait();
            }
        }
        return this.getResponseAfterDone();
    }

    @Override
    public Object get(long timeout) throws TimeoutException, RemotingException,
            InterruptedException {
        if (timeout < 0L) {
            synchronized (this) {
                while (!this.isDone) {
                    this.wait();
                }
            }
        } else if (timeout > 0L) {
            long end = System.currentTimeMillis() + timeout;
            long waitTime = timeout;
            synchronized (this) {
                while (!this.isDone && waitTime > 0L) {
                    this.wait(waitTime);
                    waitTime = end - System.currentTimeMillis();
                }
            }
        }
        return !this.isDone ? ResponseFuture.ASYN_NOT_DONE : this.getResponseAfterDone();
    }

    private Object getResponseAfterDone() throws RemotingException {
        int errorCode = this.connResponse.getResult();
        switch (errorCode) {
            case 0:
                return this.connResponse.getAppResponse();
            case 1:
            case 3:
                default:
                    throw new RemotingException(this.connResponse.getErrorMsg());
            case 2:
                String log = LogResources.getLog(RESP_TIMEOUT);
                throw new TimeoutException(log);
            case 4:
                throw new RemotingException.WriteOverFlowException(this.connResponse.getErrorMsg(),
                        (OverFlowWriteFuture) this.connResponse.getErrorCause());
        }
    }

    public void setResponse(ConnectionResponse response) {
        synchronized (this) {
            this.connResponse = response;
            this.isDone = true;
            this.notifyAll();
        }

        if (null != this.timeoutFuture) {
            this.timeoutFuture.cancel(true);
        }

        if (null != respCallback) {
            this.respCallback.getExecutor().execute(new CallBizCallback());
        }

        this.logIfResponseError(response);
    }

    private void logIfResponseError(ConnectionResponse response) {
        if (0 != response.getResult()) {
            try {
                String errorMsg = response.getErrorMsg();
                String errorStack = response.getErrorStack();
                long requestId = this.connRequest.getId();
                Object appRequest = this.connRequest.getAppRequest();
                String client = this.connRequest.getClient().toString();
                String msg;
                if (StringUtils.isBlank(errorStack)) {
                    msg = LogResources.getLog(CONNECTION_RESP_ERROR,requestId,client,appRequest,
                            errorMsg);
                } else {
                    msg = LogResources.getLog(CONNECTION_RESP_ERROR_STACK,requestId,client,
                            appRequest,errorMsg,errorStack);
                }
                LoggerInit.LOGGER.error(msg);
            } catch (Exception e) {
                LoggerInit.LOGGER.error(e.getMessage(),e);
            }
        }
    }

    public IoFutureListener getWriteFutureListener() {
        return writeFutureListener;
    }

    public void setRespCallback(ResponseCallback respCallback) {
        this.respCallback = respCallback;
    }

    private class CallBizCallback implements Runnable {

        final Thread createThread = Thread.currentThread();

        @Override
        public void run() {
            if (this.createThread == Thread.currentThread() &&
                    DefaultRespFuture.this.respCallback.getExecutor() != DIYExecutor.getInstance()) {
                String msg = LogResources.getLog(IO_THREAD_CANNOT_CALLBACK,
                        DefaultRespFuture.this.connResponse.getRequestId());
                LoggerInit.LOGGER.info(msg);
            } else {
                if (DefaultRespFuture.this.connResponse.getResult() == 0) {
                    DefaultRespFuture.this.respCallback.handleResponse(DefaultRespFuture.
                            this.connResponse.getAppResponse());
                } else if (DefaultRespFuture.this.respCallback instanceof ResponseCallback.ResponseCallbackExtends) {
                    ((ResponseCallback.ResponseCallbackExtends)(DefaultRespFuture.this.respCallback)).
                            onRemotingException(DefaultRespFuture.this.connResponse.getResult(),
                                    DefaultRespFuture.this.connResponse.getErrorMsg(),
                                    DefaultRespFuture.this.connResponse.getErrorCause());
                } else {
                    DefaultRespFuture.this.respCallback.onRemotingException(
                            DefaultRespFuture.this.connResponse.getResult(),
                            DefaultRespFuture.this.connResponse.getErrorMsg());
                }
            }
        }
    }

    private class WFL implements IoFutureListener {

        private WFL() {

        }

        @Override
        public void operationComplete(IoFuture future) {
            WriteFuture wf = UnsafeCast.cast(future);
            if (wf.isWritten()) {
                DefaultClient client = (DefaultClient) DefaultRespFuture.this.connRequest.getClient();
                DefaultRespFuture.this.waitBegin = System.currentTimeMillis();
                client.addPendingRequest(DefaultRespFuture.this.connRequest);
                TimeoutHandle timeoutHandle = new TimeoutHandle();
                DefaultRespFuture.this.timeoutFuture =
                        DefaultClientManager.timer.schedule(timeoutHandle,
                                DefaultRespFuture.this.connRequest.getRespTimeout(), TimeUnit.MILLISECONDS);
            } else if (DefaultRespFuture.this.connRequest.getDirection() != 1) {
                byte protocol = DefaultRespFuture.this.connRequest.getSerializeProtocol();
                ConnectionResponse resp = new ConnectionResponse();
                resp.setRequestId(DefaultRespFuture.this.connRequest.getId());
                resp.setSerializeProtocol(protocol);
                resp.setResult(this.getErrorCode(wf));
                resp.setErrorMsg(LogResources.getLog(REQUEST_SEND_FAIL,
                        TRConstants.getResultCodeMsg(resp.getResult())));
                resp.setErrorCause(wf);
                DefaultRespFuture.this.setResponse(resp);
            }
        }

        private int getErrorCode(WriteFuture wf) {
            return wf instanceof OverFlowWriteFuture ? 4 : 1;
        }
    }

    private class TimeoutHandle implements Callable<Boolean> {

        @Override
        public Boolean call() throws Exception {
            Client client = DefaultRespFuture.this.connRequest.getClient();
            ConnectionResponse resp = new ConnectionResponse();
            long waitMs = System.currentTimeMillis() - DefaultRespFuture.this.waitBegin;
            resp.setRequestId(DefaultRespFuture.this.connRequest.getId());
            resp.setResult(2);
            resp.setErrorMsg("响应超时[" + waitMs + "]ms.");
            return client.putResponse(resp);
        }
    }

}
